package com.jorado.redis.extend;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.*;
import redis.clients.jedis.exceptions.JedisConnectionException;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 *
 */
public class JedisClusterConnectionHandlerWraper extends JedisSlotBasedConnectionHandler {
    //
    public static Logger LOGGER = LoggerFactory.getLogger(JedisClusterConnectionHandlerWraper.class);
    //
    private final Map<String, JoradoRedisNodeInfo> nodeInfoCache = new ConcurrentHashMap<>();
    //
    private final JoradoRedisNodeInfo[] nodeSlotMappingCache = new JoradoRedisNodeInfo[BinaryJedisCluster.HASHSLOTS];
    //
    private AtomicBoolean needRefreshSlot = new AtomicBoolean(false);

    //
    private Thread redisCheckerThread = null;
    //
    private volatile boolean stopWatcherThread = false;

    private int initRefreshSlotInterval = 3000;
    //
    private volatile int refreshSlotInterval = 3000;

    public JedisClusterConnectionHandlerWraper(Set<HostAndPort> nodes,
                                               final GenericObjectPoolConfig poolConfig, int timeout) {
        this(nodes, poolConfig, timeout, timeout);

    }

    public JedisClusterConnectionHandlerWraper(Set<HostAndPort> nodes, final GenericObjectPoolConfig poolConfig,
                                               int connectionTimeout, int soTimeout) {
        super(nodes, poolConfig, connectionTimeout, soTimeout, null);
        startRefreshThread();
    }

    public JedisClusterConnectionHandlerWraper(Set<HostAndPort> nodes, GenericObjectPoolConfig poolConfig,
                                               int connectionTimeout, int soTimeout, String password) {
        super(nodes, poolConfig, connectionTimeout, soTimeout, password);
        startRefreshThread();
    }

    public JedisClusterConnectionHandlerWraper(Set<HostAndPort> nodes, GenericObjectPoolConfig poolConfig, int connectionTimeout,
                                               int soTimeout, String password, String clientName) {
        super(nodes, poolConfig, connectionTimeout, soTimeout, password, clientName);
        startRefreshThread();
    }

    protected void startRefreshThread() {
        this.initRefreshSlotInterval = NumberUtils.toInt(System.getProperty("redis.node.checker.interval", "3000"), 3000);
        this.refreshSlotInterval = this.initRefreshSlotInterval;
        this.redisCheckerThread = new Thread(() -> {
            while (true) {
                try {
                    discoverClusterNodes();
                    if (isNeedRefreshSlot()) {
                        renewSlotCache();
                        setNeedRefreshSlot(false);
                    }
                    if (!isAllNodeOk()) {
                        JedisClusterConnectionHandlerWraper.this.refreshSlotInterval = 50;
                    } else {
                        JedisClusterConnectionHandlerWraper.this.refreshSlotInterval = this.initRefreshSlotInterval;
                    }
                } catch (Throwable cause) {
                    JedisClusterConnectionHandlerWraper.this.refreshSlotInterval = 50;
                    LOGGER.warn("redis check {} error: ", cause);
                }
                try {
                    if (!stopWatcherThread) {
                        Thread.sleep(refreshSlotInterval);
                    } else {
                        break;
                    }
                } catch (Throwable cause) {
                    //
                }
            }
        });
        this.redisCheckerThread.setName("redis-state-checker");
        this.redisCheckerThread.setDaemon(true);
        this.redisCheckerThread.start();
    }

    public Jedis getConnectionFromSlot(int slot) {
        JoradoRedisNodeInfo nodeInfo = getNodeInfoBySlot(slot);
        if (nodeInfo != null && nodeInfo.isNodeFail()) {
            setNeedRefreshSlot(true);
            throw new IllegalStateException("节点" + nodeInfo.getHost() + ":" + nodeInfo.getPort() + "节点处于异常状态");
        }
        return super.getConnectionFromSlot(slot);
    }

    public void renewSlotCache(Jedis jedis) {
        super.renewSlotCache(jedis);
    }

    public void close() {
        this.stopWatcherThread = true;
        try {
            this.redisCheckerThread.interrupt();
        } catch (Throwable cause) {
            //
        }
        super.close();
    }

    /**
     * 根据槽位获取节点的信息
     *
     * @param slot 槽位值
     * @return
     */
    public JoradoRedisNodeInfo getNodeInfoBySlot(final int slot) {
        if (this.nodeSlotMappingCache != null && slot < BinaryJedisCluster.HASHSLOTS) {
            return this.nodeSlotMappingCache[slot];
        }
        return null;
    }

    public void discoverClusterNodes() {
        for (JedisPool jedisPool : this.cache.getShuffledNodesPool()) {
            Jedis jedis = null;
            try {
                jedis = jedisPool.getResource();
                discoverClusterNodes(jedis);
                return;
            } catch (JedisConnectionException exp) {
                LOGGER.warn("discoverClusterNodes error", exp);
            } finally {
                if (jedis != null) {
                    jedis.close();
                }
            }

        }
    }

    public void discoverClusterNodes(Jedis jedis) {
        String result = jedis.clusterNodes();
        if (StringUtils.isNotBlank(result)) {
            this.nodeInfoCache.clear();
            String[] clusterNodes = StringUtils.split(result, "\n");
            for (final String clusterNode : clusterNodes) {
                String[] parts = StringUtils.split(clusterNode, " ");
                if (parts.length >= 9) {
                    JoradoRedisNodeInfo nodeInfo = new JoradoRedisNodeInfo();
                    JoradoNodeState nodeState = JoradoNodeState.findNodeStateByName(parts[2]);
                    nodeInfo.setNodeState(nodeState);
                    String[] hostPort = StringUtils.split(parts[1], ":");
                    nodeInfo.setHost(hostPort[0]);
                    if (hostPort.length >= 2) {
                        nodeInfo.setPort(NumberUtils.toInt(hostPort[1], 6379));
                    }
                    String[] startEndSlot = StringUtils.split(parts[8], "-");
                    if (startEndSlot.length >= 2) {
                        nodeInfo.setStartSlot(NumberUtils.toInt(startEndSlot[0], 0));
                        nodeInfo.setEndSlot(NumberUtils.toInt(startEndSlot[1], 0));
                    }
                    nodeInfoCache.put(nodeInfo.getHost() + ":" + nodeInfo.getPort(), nodeInfo);
                }
            }
        }
        for (Map.Entry<String, JoradoRedisNodeInfo> entry : this.nodeInfoCache.entrySet()) {
            if (entry.getValue() != null) {
                for (int solt = entry.getValue().getStartSlot(); solt <= entry.getValue().getEndSlot(); ++solt) {
                    if (solt < BinaryJedisCluster.HASHSLOTS) {
                        this.nodeSlotMappingCache[solt] = entry.getValue();
                    }
                }
            }
        }
    }

    public void setNeedRefreshSlot(boolean flag) {
        this.needRefreshSlot.set(flag);
    }

    public boolean isNeedRefreshSlot() {
        return this.needRefreshSlot.get();
    }

    private boolean isAllNodeOk() {
        for (Map.Entry<String, JoradoRedisNodeInfo> entry : this.nodeInfoCache.entrySet()) {
            if (entry.getValue() != null && entry.getValue().isNodeFail()) {
                return false;
            }
        }
        return true;
    }

}
